package cdrprocessing.topology;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.Map.Entry;
import java.util.Properties;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.BoltDeclarer;
import backtype.storm.topology.TopologyBuilder;
import cdrprocessing.bolt.filter.BloomFilterBolt;
import cdrprocessing.bolt.filter.FilterMtcTupleBolt;
import cdrprocessing.bolt.rule.ExportResultBolt;
import cdrprocessing.bolt.rule.RuleEvaluatorBolt;
import cdrprocessing.bolt.rule.RuleLoadBalancingBolt;
import cdrprocessing.bolt.stat.CallPerSecBolt;
import cdrprocessing.bolt.stat.DurationPerSecBolt;
import cdrprocessing.bolt.stat.ExportStatBolt;
import cdrprocessing.bolt.stat.FailCallPerSecBolt;
import cdrprocessing.bolt.util.ReduceFieldBolt;
import cdrprocessing.bolt.util.ThrottlingBolt;
import cdrprocessing.spout.RawCdrFileSpout;
import cdrprocessing.spout.ReadRuleFromSocketSpout;

public class CdrProcessingApp {

    public static void main(String[] args) {
        final String configFile = "src/main/resources/config.properties";

        final Properties properties = new Properties();
        try {
            properties.load(new FileInputStream(configFile));
        } catch (IOException ex) {
            System.out.println("Properties file not found");
            System.exit(-1);
        }

        Config defaultConfig = new Config();
        for (final Entry<Object, Object> property : properties.entrySet()) {
            defaultConfig.put((String) property.getKey(), property.getValue());
        }
        defaultConfig.setDebug(false);

        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("rawCdrSpout", new RawCdrFileSpout());
        builder.setBolt("filterMtcTupleBolt", new FilterMtcTupleBolt()).shuffleGrouping("rawCdrSpout", "RawCdrStream");
        //builder.setBolt("filterDuplicateTupleBolt", new BloomFilterBolt()).shuffleGrouping("filterMtcTupleBolt", "RawCdrStream");
        builder.setBolt("reduceFieldBolt", new ReduceFieldBolt()).shuffleGrouping("filterMtcTupleBolt", "RawCdrStream");
        builder.setBolt("throttlingBolt", new ThrottlingBolt()).shuffleGrouping("reduceFieldBolt", "CdrStream");

        builder.setBolt("callPerSecBolt", new CallPerSecBolt()).shuffleGrouping("throttlingBolt", "CdrStream");
        builder.setBolt("durationPerSecBolt", new DurationPerSecBolt()).shuffleGrouping("throttlingBolt", "CdrStream");
        builder.setBolt("failCallPerSecBolt", new FailCallPerSecBolt()).shuffleGrouping("throttlingBolt", "CdrStream");
        builder.setBolt("exportStatBolt", new ExportStatBolt()).shuffleGrouping("callPerSecBolt", "CallPerSec")
                .shuffleGrouping("durationPerSecBolt", "DurationPerSec").shuffleGrouping("failCallPerSecBolt", "FailCallPerSec");

        builder.setSpout("ruleSpout", new ReadRuleFromSocketSpout());
        builder.setBolt("ruleBalancer", new RuleLoadBalancingBolt()).shuffleGrouping("ruleSpout", "RuleStream");
        for (int i = 0; i < Integer.valueOf(properties.getProperty("numberOfPartitions")); i++) {
            builder.setBolt("RuleEvaluator" + i, new RuleEvaluatorBolt()).shuffleGrouping("ruleBalancer", "RuleStream" + i)
                    .shuffleGrouping("throttlingBolt", "CdrStream");
        }
        final BoltDeclarer exportResultBolt = builder.setBolt("exportResultBolt", new ExportResultBolt());
        for (int i = 0; i < Integer.valueOf(properties.getProperty("numberOfPartitions")); i++) {
            exportResultBolt.shuffleGrouping("RuleEvaluator" + i, "DiscountForPopularUsersRule")
                    .shuffleGrouping("RuleEvaluator" + i, "DiscountForManyCallsRule").shuffleGrouping("RuleEvaluator" + i, "DiscountForFailureRule");
        }

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("cdr-process", defaultConfig, builder.createTopology());
    }
}
